/*
 * Copyright 2008 Konstantin Triger <kostat@gmail.com> 
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * You may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *  http://www.apache.org/licenses/LICENSE-2.0 
 *  
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package jaque;

import jaque.functions.*;
import jaque.misc.*;
import jaque.functions.math.BinaryOperator;

import java.util.Collections;
import java.util.Iterator;

import static jaque.functions.Expression.*;

/**
 * Provides a comprehensive set of methods receiving and returning object
 * streams. Each method applies a business rule in a specific way and returns a
 * new object stream for further processing.
 * 
 * A very useful application is feeding one function with output of another
 * function, i.e.<p/>
 * 
 * {@code select(where(...), ...). }
 * 
 * @author <a href="mailto://object_streaming@googlegroups.com">Konstantin
 *         Triger</a>
 */
public final class Query {

	private Query() {
	}

	/**
	 * Concatenates two object streams. See also
	 * {@link Query#union(Iterable, Iterable)}.
	 * 
	 * @param <T>
	 *            the type of the objects in the streams.
	 * @param first
	 *            the first stream.
	 * @param second
	 *            the second stream.
	 * @return the concatenated stream.
	 */
	public static <T> Iterable<T> concat(final Iterable<T> first,
			final Iterable<T> second) {

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return new BaseIterator<T>() {
					int _state;
					Iterator<T> _iter;

					// @Override
					public final boolean hasNext() {

						switch (_state) {
						case 0:
							_iter = first.iterator();
							_state = 1;
						case 1:
							if (_iter.hasNext())
								return yield(_iter.next());

							_iter = second.iterator();
							_state = 2;
						case 2:
							if (_iter.hasNext())
								return yield(_iter.next());
						}

						return yieldBreak;
					}
				};
			}
		};
	}

	/**
	 * Returns an empty object stream of a specific type.
	 * 
	 * @param <T>
	 *            the type of the objects in the stream.
	 * @return an empty object stream of a specific type.
	 */
	public static <T> Iterable<T> empty() {
		return Collections.<T> emptyList();
	}

	/**
	 * Correlates the objects of two streams based on matching keys.
	 * 
	 * <pre>
	 * Object.equals()
	 * </pre>
	 * 
	 * is used to compare keys.
	 * 
	 * @param <T>
	 *            the type of the objects in the first (outer) stream.
	 * @param <R>
	 *            the type of the objects in the second (inner) stream.
	 * @param <Key>
	 *            The type of the keys returned by the key selector functions.
	 * @param <Result>
	 *            the type of the object in the resulting stream.
	 * @param outer
	 *            the first stream to join.
	 * @param outerSelector
	 *            a function to extract the join key from each object of the
	 *            first stream.
	 * @param inner
	 *            the second stream to join.
	 * @param innerSelector
	 *            a function to extract the join key from each object of the
	 *            second stream.
	 * @param resultSelector
	 *            a function to create a result object from two matching
	 *            objects.
	 * @return an object stream containing objects obtained by performing an
	 *         inner join on two streams.
	 */
	public static <T, R, Key, Result> Iterable<Result> join(
			final Iterable<T> outer,
			final Function<? extends Key, ? super T> outerSelector,
			final Iterable<R> inner,
			final Function<? extends Key, ? super R> innerSelector,
			final Function2<Result, ? super T, ? super R> resultSelector) {
		return join(outer, outerSelector, inner, innerSelector, resultSelector,
				null);
	}

	/**
	 * Correlates the objects of two streams based on matching keys. A specified
	 * comparator is used to compare keys.
	 * 
	 * @param <T>
	 *            the type of the objects in the first (outer) stream.
	 * @param <R>
	 *            the type of the objects in the second (inner) stream.
	 * @param <Key>
	 *            The type of the keys returned by the key selector functions.
	 * @param <Result>
	 *            the type of the object in the resulting stream.
	 * @param outer
	 *            the first stream to join.
	 * @param outerSelector
	 *            a function to extract the join key from each object of the
	 *            first stream.
	 * @param inner
	 *            the second stream to join.
	 * @param innerSelector
	 *            a function to extract the join key from each object of the
	 *            second stream.
	 * @param resultSelector
	 *            a function to create a result object from two matching
	 *            objects.
	 * @param comparator
	 *            comparator is used to compare keys.
	 * @return an object stream containing objects obtained by performing an
	 *         inner join on two streams.
	 */
	public static <T, R, Key, Result> Iterable<Result> join(
			final Iterable<T> outer,
			final Function<? extends Key, ? super T> outerSelector,
			final Iterable<R> inner,
			final Function<? extends Key, ? super R> innerSelector,
			final Function2<Result, ? super T, ? super R> resultSelector,
			final java.util.Comparator<? super Key> comparator) {

		final java.util.Comparator<? super Key> comp = Utils
				.<java.util.Comparator<? super Key>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {
				return new BaseIterator<Result>() {

					int _state;
					Iterator<T> _outer;
					Iterator<R> _inner;
					T _currentOuter;
					Key _outerKey;

					// @Override
					public final boolean hasNext() {

						switch (_state) {
						case 0:
							_outer = outer.iterator();
							_state = 1;
						case 1:
							if (!_outer.hasNext())
								break;

							_currentOuter = _outer.next();
							_inner = inner.iterator();
							_outerKey = null;
							_state = 2;
						case 2:
							try {
								_outerKey = outerSelector.invoke(_currentOuter);
							} catch (RuntimeException re) {
								throw re;
							} catch (Throwable e) {
								throw new RuntimeException(e);
							}
							_state = 3;
						case 3:
							while (_inner.hasNext()) {
								R current = _inner.next();
								try {
									if (comp.compare(_outerKey, innerSelector
											.invoke(current)) == 0)
										return yield(resultSelector.invoke(
												_currentOuter, current));
								} catch (RuntimeException re) {
									throw re;
								} catch (Throwable e) {
									throw new RuntimeException(e);
								}
							}

							_state = 1;
							return hasNext();
						}

						return yieldBreak;
					}
				};
			}
		};
	}

	/**
	 * Transforms each object in the stream into another object.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param <Result>
	 *            type of the objects in the output stream.
	 * @param source
	 *            the input object stream.
	 * @param transform
	 *            a function to transform input object into output object.
	 * @return an object stream containing transformed objects.
	 */
	public static <T, Result> Iterable<Result> select(final Iterable<T> source,
			final Function<? extends Result, ? super T> transform) {
		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {

				return new BaseIterator<Result>() {
					final private Iterator<T> _iter = source.iterator();

					// @Override
					public final boolean hasNext() {
						try {
							if (_iter.hasNext())
								return yield(transform.invoke(_iter.next()));
						} catch (RuntimeException re) {
							throw re;
						} catch (Throwable e) {
							throw new RuntimeException(e);
						}

						return yieldBreak;
					}
				};
			}
		};

	}

	/**
	 * Recursively aggregates each object in the input stream with the
	 * aggregated result of previous objects. e.g.
	 * 
	 * Sum(n objects) == Sum(n-1 objects) + n<sup>th</sup> object.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param <Result>
	 *            type of the objects in the output stream.
	 * @param source
	 *            the input object stream.
	 * @param aggregator
	 *            a function to combine an aggregated result of the sequence
	 *            preceding current object with the current object.
	 * @return an object stream containing aggregated results for each object in
	 *         the input stream.
	 */
	public static <T, Result> Iterable<Result> aggregate(
			final Iterable<T> source,
			final Function2<? extends Result, ? super Result, ? super T> aggregator) {
		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {

				return new BaseIterator<Result>() {
					private final Iterator<T> _iter = source.iterator();

					// @Override
					public final boolean hasNext() {
						if (_iter.hasNext()) {
							try {
								return yield(aggregator.invoke(next(), _iter
										.next()));
							} catch (RuntimeException re) {
								throw re;
							} catch (Throwable e) {
								throw new RuntimeException(e);
							}
						}

						return yieldBreak;
					}
				};
			}
		};

	}

	/**
	 * Computes the average of a stream of numeric objects.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @return the average.
	 */
	public static <T extends Number> Iterable<Number> average(
			final Iterable<T> source) {
		return average(source, Expression.<T> self());
	}

	/**
	 * Computes the average of a stream of objects by applying a transform to
	 * numeric to each element.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @return the average.
	 */
	public static <T, E extends Number> Iterable<Number> average(
			final Iterable<T> source, final Function<E, ? super T> selector) {
		final Iterable<E> aggregate = sum(source, selector);

		return new Iterable<Number>() {
			public final Iterator<Number> iterator() {

				return new BaseIterator<Number>() {
					private final Iterator<E> _iter = aggregate.iterator();
					long _count;

					// @Override
					public final boolean hasNext() {

						while (_iter.hasNext()) {
							_count++;
							return yield(_iter.next());
						}

						return yieldBreak;
					}

					@Override
					public final Number next() {
						return BinaryOperator.Divide.eval(super.next(), _count);
					}
				};
			}
		};
	}

	/**
	 * Returns a maximum value in a stream of numeric objects.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @return the maximum value.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> max(
			final Iterable<T> source) {
		return max(source, Expression.<T> self(), Comparator
				.<T> getComparableDefault());
	}

	/**
	 * Computes the maximum value in a stream of objects by transforming each
	 * element to numeric value. *
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @return the maximum value.
	 */
	public static <T, Key extends Comparable<? super Key>> Iterable<T> max(
			final Iterable<T> source,
			final Function<? extends Key, ? super T> selector) {
		return max(source, selector, Comparator.<Key> getComparableDefault());
	}

	/**
	 * Computes the maximum value in a stream of objects by transforming each
	 * element to numeric value. A specified comparator is used to compare
	 * values.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return the maximum value.
	 */
	public static <T, Key> Iterable<T> max(final Iterable<T> source,
			final Function<? extends Key, ? super T> selector,
			final java.util.Comparator<? super Key> comparator) {
		return aggregate(source, conditional(Expression
				.<T, Key> greaterThanOrEqual(selector, selector, comparator)));
	}

	/**
	 * Returns a minimum value in a stream of numeric objects.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @return the minimum value.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> min(
			final Iterable<T> source) {
		return min(source, Expression.<T> self(), Comparator
				.<T> getComparableDefault());
	}

	/**
	 * Computes the minimum value in a stream of objects by transforming each
	 * element to numeric value.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @return the minimum value.
	 */
	public static <T, Key extends Comparable<? super Key>> Iterable<T> min(
			final Iterable<T> source,
			final Function<? extends Key, ? super T> selector) {
		return min(source, selector, Comparator.<Key> getComparableDefault());
	}

	/**
	 * Computes the minimum value in a stream of objects by transforming each
	 * element to numeric value. A specified comparator is used to compare
	 * values.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return the minimum value.
	 */
	public static <T, Key> Iterable<T> min(final Iterable<T> source,
			final Function<? extends Key, ? super T> selector,
			final java.util.Comparator<? super Key> comparator) {
		return aggregate(source, conditional(Expression
				.<T, Key> lessThanOrEqual(selector, selector, comparator)));
	}

	/**
	 * Computes the sum of a stream of objects by transforming each element to
	 * numeric value.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param selector
	 *            a transform function to apply to each input object.
	 * @return the sum value.
	 */
	public static <T, Result extends Number> Iterable<Result> sum(
			final Iterable<T> source, final Function<Result, ? super T> selector) {
		return aggregate(source, Expression.<Result, T> add(selector, selector));
	}

	/**
	 * Returns the sum of a stream of numeric objects.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @return the sum value.
	 */
	public static <T extends Number> Iterable<T> sum(final Iterable<T> source) {
		return sum(source, Expression.<T> self());
	}

	/**
	 * Returns distinct elements from an object stream.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @return an object stream containing distinct elements.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> distinct(
			final Iterable<T> source) {
		return distinct(source, Comparator.<T> getComparableDefault());
	}

	/**
	 * Returns distinct elements from an object stream. A specified comparator
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            type of the objects in the input stream.
	 * @param source
	 *            the input object stream.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return an object stream containing distinct elements.
	 */
	public static <T> Iterable<T> distinct(final Iterable<T> source,
			final java.util.Comparator<? super T> comparator) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return new BaseIterator<T>() {
					int _state;
					Iterator<T> _iter;
					java.util.Set<T> _set;

					// @Override
					public final boolean hasNext() {

						switch (_state) {
						case 0:
							_set = new java.util.TreeSet<T>(comp);
							_iter = source.iterator();
							_state = 1;
						case 1:
							while (_iter.hasNext()) {
								T current = _iter.next();
								if (_set.add(current))
									return yield(current);
							}
						}

						return yieldBreak;
					}
				};
			}
		};
	}

	/**
	 * Produces the intersection of two object streams.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the streams.
	 * @param first
	 *            the first stream.
	 * @param second
	 *            the second stream.
	 * @return an object stream forming the intersection of two input streams.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> intersect(
			final Iterable<T> first, final Iterable<T> second) {
		return intersect(first, second, Comparator.<T> getComparableDefault());
	}

	/**
	 * Produces the intersection of two object streams. A specified comparator
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the streams.
	 * @param first
	 *            the first stream.
	 * @param second
	 *            the second stream.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return an object stream forming the intersection of two input streams.
	 */
	public static <T> Iterable<T> intersect(final Iterable<T> first,
			final Iterable<T> second,
			final java.util.Comparator<? super T> comparator) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return new BaseIterator<T>() {
					int _state;
					Iterator<T> _iter;
					java.util.Set<T> _set;

					// @Override
					public final boolean hasNext() {

						switch (_state) {
						case 0:
							_set = new java.util.TreeSet<T>(comp);
							for (T t : second)
								_set.add(t);
							_iter = first.iterator();
							_state = 1;
						case 1:
							while (_iter.hasNext()) {
								T current = _iter.next();
								if (_set.remove(current))
									return yield(current);
							}
						}

						return yieldBreak;
					}
				};
			}
		};
	}

	/**
	 * Concatenates two object streams, excluding duplicates. A specified
	 * comparator is used to compare values. See also
	 * {@link Query#concat(Iterable, Iterable)}.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the streams.
	 * @param first
	 *            the first stream.
	 * @param second
	 *            the second stream.
	 * @return the concatenated stream.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> union(
			final Iterable<T> first, final Iterable<T> second) {
		return union(first, second, Comparator.<T> getComparableDefault());
	}

	/**
	 * Concatenates two object streams, excluding duplicates. See also
	 * {@link Query#concat(Iterable, Iterable)}.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the streams.
	 * @param first
	 *            the first stream.
	 * @param second
	 *            the second stream.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return the concatenated stream.
	 */
	public static <T> Iterable<T> union(final Iterable<T> first,
			final Iterable<T> second,
			final java.util.Comparator<? super T> comparator) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return new BaseIterator<T>() {
					int _state;
					Iterator<T> _iter;
					java.util.Set<T> _set;

					// @Override
					public final boolean hasNext() {

						switch (_state) {
						case 0:
							_iter = first.iterator();
							_set = new java.util.TreeSet<T>(comp);
							_state = 1;
						case 1:
							while (_iter.hasNext()) {
								T current = _iter.next();
								if (_set.add(current))
									return yield(current);
							}

							_iter = second.iterator();
							_state = 2;
						case 2:
							while (_iter.hasNext()) {
								T current = _iter.next();
								if (_set.add(current))
									return yield(current);
							}
						}

						return yieldBreak;
					}
				};
			}
		};
	}

	/**
	 * Generates an object stream consisting of {@code count} copies of the
	 * specified object.
	 * 
	 * @param <T>
	 *            the type of the object.
	 * @param element
	 *            an element to be repeated.
	 * @param count
	 *            number of times to repeat the element.
	 * @return an object stream consisting of {@code count} copies of the
	 *         specified object stream.
	 */
	public static <T> Iterable<T> repeat(final T element, final int count) {
		// return Collections.nCopies(count, element);
		return new Iterable<T>() {
			public final Iterator<T> iterator() {

				return new BaseIterator<T>() {
					{
						yield(element);
					}

					int _count = count;

					// @Override
					public final boolean hasNext() {
						return _count-- > 0;
					}
				};
			}
		};
	}

	/**
	 * Bypasses a specified number of elements in a object stream and then
	 * returns the remaining ones.
	 * 
	 * @param <T>
	 *            the type of the object.
	 * @param source
	 *            the input object stream.
	 * @param count
	 *            number of elements to skip.
	 * @return an object stream consisting of remaining objects.
	 */
	public static <T> Iterable<T> skip(final Iterable<T> source, final int count) {
		return where(source, indexGreaterThanOrEqual(count));
	}

	/**
	 * Returns a specified number of elements from start of the object stream.
	 * 
	 * @param <T>
	 *            the type of the object.
	 * @param source
	 *            the input object stream.
	 * @param count
	 *            number of elements to return.
	 * @return an object stream containing the specified number of the objects
	 *         from beginning of the input stream.
	 */
	public static <T> Iterable<T> take(final Iterable<T> source, final int count) {
		return where(source, indexLessThan(count));
	}

	/**
	 * Filters the input object stream according with the specified predicate.
	 * 
	 * @param <E>
	 *            the type of the objects in the stream.
	 * @param source
	 *            the input object stream.
	 * @param predicate
	 *            A function to test each object in the stream.
	 * @return filtered object stream.
	 */
	public static <E> Iterable<E> where(final Iterable<E> source,
			final Predicate<? super E> predicate) { // Comparable<T>
		return new Iterable<E>() {
			public final Iterator<E> iterator() {

				return new BaseIterator<E>() {
					private final Iterator<E> _iter = source.iterator();

					// @Override
					public final boolean hasNext() {

						while (_iter.hasNext()) {
							E current = _iter.next();
							try {
								if (predicate.invoke(current))
									return yield(current);
							} catch (RuntimeException re) {
								throw re;
							} catch (Throwable e) {
								throw new RuntimeException(e);
							}
						}

						return yieldBreak;
					}
				};
			}
		};
	}
}
